package com.summerbird.mallchat.common.websocket.service.impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.RandomUtil;
import cn.hutool.json.JSONUtil;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.summerbird.mallchat.common.common.constant.RedisKey;
import com.summerbird.mallchat.common.common.event.UserOfflineEvent;
import com.summerbird.mallchat.common.common.event.UserOnlineEvent;
import com.summerbird.mallchat.common.common.utils.RedisUtils;
import com.summerbird.mallchat.common.user.dao.UserDao;
import com.summerbird.mallchat.common.user.domain.entity.User;
import com.summerbird.mallchat.common.user.service.LoginService;
import com.summerbird.mallchat.common.user.service.impl.WXMsgServiceImpl;
import com.summerbird.mallchat.common.websocket.NettyUtil;
import com.summerbird.mallchat.common.websocket.domain.dto.WSChannelExtraDTO;
import com.summerbird.mallchat.common.websocket.domain.vo.response.WSBaseResp;
import com.summerbird.mallchat.common.websocket.service.WebSocketService;
import com.summerbird.mallchat.common.websocket.service.adapter.WebSocketAdapter;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import me.chanjar.weixin.mp.api.WxMpService;
import me.chanjar.weixin.mp.bean.result.WxMpQrCodeTicket;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.Date;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

import static com.summerbird.mallchat.common.common.config.ThreadPoolConfig.WS_EXECUTOR;

@Slf4j
@Service
public class WebSocketServiceImpl implements WebSocketService {
    private static final Duration EXPIRE_TIME = Duration.ofHours(1);
    /**
     * redis保存loginCode的key
     */
    private static final String LOGIN_CODE = "loginCode";

    @Autowired
    @Lazy
    private WxMpService wxMpService;
    @Autowired
    private UserDao userDao;
    @Autowired
    private LoginService loginService;
    @Autowired
    @Lazy
    private WXMsgServiceImpl wxMsgService;
    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;
    @Autowired
    @Qualifier(WS_EXECUTOR)
    private ThreadPoolTaskExecutor executor;
    /**
     * 管理所有用户连接（登录态/游客）
     * channel 与 uid
     */
    public static final ConcurrentHashMap<Channel, WSChannelExtraDTO> ONLINE_WS_MAP = new ConcurrentHashMap<>();
    /**
     * 所有在线的用户和对应的socket
     */
    private static final ConcurrentHashMap<Long, CopyOnWriteArrayList<Channel>> ONLINE_UID_MAP = new ConcurrentHashMap<>();

    public static final int MAXIMUM_SIZE = 1000;
    public static final Duration EXPIRETIME = Duration.ofHours(1);

    /**
     * 管理登陆码code与channel的关系
     */
    public static final Cache<Integer, Channel> WAIT_LOGIN_MAP = Caffeine.newBuilder()
            .maximumSize(MAXIMUM_SIZE)
            .expireAfterWrite(EXPIRETIME)
            .build();

    @Override
    public void connect(Channel channel) {
        ONLINE_WS_MAP.put(channel, new WSChannelExtraDTO());
    }

    @SneakyThrows
    @Override
    public void handleLoginReq(Channel channel) {
         // 1. 生成随机登录码
        Integer code = generateLoginCode(channel);
        // 2. 生成临时二维码
        WxMpQrCodeTicket wxMpQrCodeTicket = wxMpService.getQrcodeService().qrCodeCreateTmpTicket(code, (int) EXPIRETIME.getSeconds());
        // 3. 推送给前端
        sendMsg(channel, WebSocketAdapter.buildLoginResp(wxMpQrCodeTicket));
    }

    @Override
    public void remove(Channel channel) {
        WSChannelExtraDTO wsChannelExtraDTO = ONLINE_WS_MAP.get(channel);
        Optional<Long> optional = Optional.ofNullable(wsChannelExtraDTO)
                .map(WSChannelExtraDTO::getUid);
        // 判断是否全部设备都下线了
        boolean isAllOffLine = offline(channel, optional);
        if(optional.isPresent() && isAllOffLine){ // 发送用户下线事件
            User user = new User();
            user.setId(optional.get());
            user.setLastOptTime(new Date());// 更新下线时间
            applicationEventPublisher.publishEvent(new UserOfflineEvent(this, user));
        }
    }

    private boolean offline(Channel channel, Optional<Long> optional) {
        // 下线当前channel
        ONLINE_WS_MAP.remove(channel);
        // 与该uid的所有channel是否全部下线？
        if(optional.isPresent()){
            Long uid = optional.get();
            CopyOnWriteArrayList<Channel> channels = ONLINE_UID_MAP.get(uid);
            if(!CollectionUtils.isEmpty(channels)){
                channels.removeIf(ch -> Objects.equals(ch, channel));
            }
            return CollectionUtils.isEmpty(channels);
        }
        return true;
    }


    @Override
    public void scanLoinSuccess(Integer code, Long uid) {
        // 确保链接在机器上
        Channel channel = WAIT_LOGIN_MAP.getIfPresent(code);
        if(Objects.isNull(channel)){
            return;
        }
        User user = userDao.getById(uid);
        // 移除code
        WAIT_LOGIN_MAP.invalidate(code);
        // 获取登录token
        String token = loginService.login(uid);
        // 发送信息给前端
        loginSuccess(channel, user, token);
    }

    /**
     * 等待授权时，推送消息给前端
     * @param code
     */
    @Override
    public void waitAuthorize(Integer code) {
        Channel channel = WAIT_LOGIN_MAP.getIfPresent(code);
        // 发送信息给前端
        sendMsg(channel, WebSocketAdapter.buildLoginScanSuccessResp());
    }

    @Override
    public void authorize(Channel channel, String token) {
        Long uid = loginService.getValidUid(token);
        if(Objects.isNull(uid)){
            sendMsg(channel, WebSocketAdapter.buildInvalidDateTokenResp());
        }else {
            User user = userDao.getById(uid);
            loginSuccess(channel, user, token);
        }
    }

    @Override
    public void sendMsgToAll(WSBaseResp<?> msg, Long skipUid) {
        ONLINE_WS_MAP.forEach((channel, ext) ->{
            if (Objects.nonNull(skipUid) && Objects.equals(ext.getUid(), skipUid)) {
                return;
            }
            executor.execute(()->{
                sendMsg(channel, msg);
            });
        });
    }

    @Override
    public void sendToUid(WSBaseResp<?> wsBaseMsg, Long uid) {
        CopyOnWriteArrayList<Channel> channels = ONLINE_UID_MAP.get(uid);
        if(CollectionUtils.isEmpty(channels)){
            log.info("用户{}不在线", uid);
            return;
        }
        channels.forEach(channel -> executor.execute(() -> sendMsg(channel, wsBaseMsg)));
    }

    @Override
    public void sendToAllOnline(WSBaseResp<?> wsBaseResp, Long skipUid) {
        ONLINE_WS_MAP.forEach((channel, ext) -> {
            if(Objects.nonNull(skipUid) && Objects.equals(ext.getUid(), skipUid)){
                return;
            }
            executor.execute(() -> sendMsg(channel, wsBaseResp));
        });
    }

    private void loginSuccess(Channel channel, User user, String token) {
        // 保存channel对应uid
        online(channel, user.getId());
        // 推送登录成功消息
        sendMsg(channel, WebSocketAdapter.buildLoginSuccessResp(user, token));
        // 用户上线成功的事件
        user.setLastOptTime(new Date());
        user.refreshIp(NettyUtil.getAttr(channel, NettyUtil.IP));
        applicationEventPublisher.publishEvent(new UserOnlineEvent(this, user));
    }

    private void online(Channel channel, Long uid) {
        WSChannelExtraDTO wsChannelExtraDTO = getOrInitChannelExt(channel);
        ONLINE_UID_MAP.putIfAbsent(uid, new CopyOnWriteArrayList<>());
        ONLINE_UID_MAP.get(uid).add(channel);
        NettyUtil.setAttr(channel, NettyUtil.UID, uid);
    }

    /**
     * 如果在线列表不存在，就先把该channel放进在线列表
     *
     * @param channel
     * @return
     */
    private WSChannelExtraDTO getOrInitChannelExt(Channel channel) {
        WSChannelExtraDTO wsChannelExtraDTO =
                ONLINE_WS_MAP.getOrDefault(channel, new WSChannelExtraDTO());
        WSChannelExtraDTO old = ONLINE_WS_MAP.putIfAbsent(channel, wsChannelExtraDTO);
        return ObjectUtil.isNull(old) ? wsChannelExtraDTO : old;
    }


    private void sendMsg(Channel channel, WSBaseResp<?> wsBaseResp) {
        channel.writeAndFlush(new TextWebSocketFrame(JSONUtil.toJsonStr(wsBaseResp)));
    }

    /**
     * 获取不重复的登录的code，微信要求最大不超过int的存储极限
     * 防止并发，可以给方法加上synchronize，也可以使用cas乐观锁
     *
     * @return
     */
    private Integer generateLoginCode(Channel channel) {
        int inc;
        do {
            //本地cache时间必须比redis key过期时间短，否则会出现并发问题
            inc = RedisUtils.integerInc(RedisKey.getKey(LOGIN_CODE), (int) EXPIRE_TIME.toMinutes(), TimeUnit.MINUTES);
        } while (WAIT_LOGIN_MAP.asMap().containsKey(inc));
        //储存一份在本地
        WAIT_LOGIN_MAP.put(inc, channel);
        return inc;
    }
}
